package redis

import (
	"context"
	"fmt"
	"time"

	"gitee.com/xwengf/go-event/v2/event"

	"github.com/go-redis/redis/v8"
)

type redisReceiver struct {
	client *redis.Client
	topic  string
}

type redisSender struct {
	client *redis.Client
}

// Close implements event.Sender.
func (r *redisSender) Close() error {
	return r.client.Close()
}

// Send implements event.Sender.
func (r *redisSender) Send(ctx context.Context, topic string, msg event.Event) error {
	// 提前判断原 ctx 是否已超时或取消
	if ctx.Err() != nil {
		fmt.Println("上下文已超时，重新定义上下文")
		// 创建新的 Redis 专用 ctx，避免误伤
		newCtx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		ctx = newCtx
	}
	// 使用 XAdd 向 Stream 中添加消息
	_, err := r.client.XAdd(ctx, &redis.XAddArgs{
		Stream: topic, // Stream 名称,
		Values: map[string]interface{}{"message": msg.Value()},
	}).Result()
	if err != nil {
		return fmt.Errorf("failed to publish message: %v", err)
	}
	return nil
}

type Message struct {
	key   string
	value []byte
}

func (m *Message) Key() string {
	return m.key
}

func (m *Message) Value() []byte {
	return m.value
}

func NewMessage(key string, value []byte) event.Event {
	return &Message{
		key:   key,
		value: value,
	}
}

// Close implements event.Receiver.
func (r *redisReceiver) Close() error {
	fmt.Println("关闭redis消费者：", r.topic)
	return r.client.Close()
}

// Receive implements event.Receiver.
func (r *redisReceiver) Receive(ctx context.Context, handler event.Handler) error {
	fmt.Println("开始监听redis队列：", r.topic)
	defer func() {
		fmt.Println("Redis stream 已退出！！")
		_ = r.client.Close()
	}()
	lastID := "0"
	for {
		select {
		case <-ctx.Done(): // 监听上下文是否取消
			fmt.Printf("[Queue] subscribe stopped!\n")
			return nil
		default:
			// 使用 XRead 读取消息
			// fmt.Printf("使用 XRead 读取消息: lastID=%s", lastID)
			result, err := r.client.XRead(ctx, &redis.XReadArgs{
				Streams: []string{r.topic, lastID},
				Block:   0, // 阻塞等待新消息
				Count:   1, // 每次读取一条消息
			}).Result()
			if err != nil {
				fmt.Printf("failed to read from stream: %v\n", err)
				return fmt.Errorf("failed to read from stream: %v", err)
			}
			// 处理读取到的消息
			for _, stream := range result {
				for _, message := range stream.Messages {
					fmt.Printf("Received: (ID: %s)", message.ID)
					// 处理消息
					str := message.Values["message"].(string)
					fmt.Printf("消息内容：%s", str)
					err := handler(ctx, r.topic, NewMessage(message.ID, []byte(str)))
					if err != nil {
						fmt.Printf("处理消息失败: %v", err)
						// return err
					}
					// 删除已处理的消息
					_, err = r.client.XDel(ctx, r.topic, message.ID).Result()
					if err != nil {
						return fmt.Errorf("failed to delete message: %v", err)
					}
					fmt.Printf("Deleted: %s\n", message.ID)
					lastID = message.ID // 更新最后读取的 ID
				}
			}
		}

	}
}

func NewRedisReceiver(addr, username, password, topic string, wd time.Duration, rd time.Duration) event.Receiver {
	rdb := redis.NewClient(&redis.Options{
		Addr:         addr,
		WriteTimeout: wd,
		ReadTimeout:  rd,
		Password:     password,
		Username:     username,
	})
	return &redisReceiver{
		client: rdb,
		topic:  topic,
	}
}

func NewRedisSender(addr, username, password string, wd time.Duration, rd time.Duration) event.Sender {
	rdb := redis.NewClient(&redis.Options{
		Addr:         addr,
		WriteTimeout: wd,
		ReadTimeout:  rd,
		Password:     password,
		Username:     username,
	})
	return &redisSender{
		client: rdb,
	}
}
